package com.shujia.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo19Action {
  def main(args: Array[String]): Unit = {
    // Action行为算子
    // foreach、foreachPartition、collect、colletAsMap、take、reduce、reduceByKeyLocally、count、top、lookup


    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo19Action")

    val sc: SparkContext = new SparkContext(conf)

    val stuRDD: RDD[String] = sc.textFile("spark/data/stu/students.txt")
    // collect：将RDD的所有数据变成本地集合Array数组，数组的泛型同RDD中的一致
    // 注意数据量的问题，本地集合有可能放不下

    // collectAsMap：将数据变成KV格式的Map
    val stuArr: Array[String] = stuRDD.collect()
    println(stuArr)
    stuArr.foreach(println)

    val clazzRDD: RDD[(String, Int)] = stuRDD.map(line => (line.split(",")(4), 1))
    val clazzArr: Array[(String, Int)] = clazzRDD.collect()
    val clazzMap: collection.Map[String, Int] = clazzRDD.collectAsMap()
    println(clazzArr)
    println(clazzMap)

    // take：取出前n条数据并变成Scala的本地集合
    stuRDD.take(10).foreach(println)

    // reduce：将全局数据进行聚合
    // 统计所有学生的年龄之和
    val sumAge: Int = stuRDD.map(_.split(",")(2).toInt)
      .groupBy(age => "1")
      .map(_._2.sum)
      .take(1)(0)

    println(sumAge)

    val sumAge02: Int = stuRDD.map(_.split(",")(2).toInt)
      .reduce(_ + _)
    println(sumAge02)

    // reduceByKeyLocally
    stuRDD.map(line => (line.split(",")(4), 1))
      .reduceByKey(_ + _)
      .foreach(println)

    stuRDD.map(line => (line.split(",")(4), 1))
      .reduceByKeyLocally(_ + _) // 同样的会进行分组聚合预聚合，并将最终的结果转换成本地结合Map
      .foreach(println) // foreach是本地集合Map提供的

    // count：统计RDD的数据量
    val cnt: Long = stuRDD.count()
    println(cnt)

    // top：将数据进行降序排列后再返回n条数据
    stuRDD.repartition(10)
      .top(10).foreach(println) // 返回前10条数据
    // take从每个分区去一部分数据出来，只能保证取出10条数据
    stuRDD.repartition(10).take(10).foreach(println) // 返回前10条数据


    // lookup：需要作用在KV格式的RDD上，可以通过Key获取所有的Value
    val seq: Seq[String] = stuRDD.map(line => (line.split(",")(0), line))
      .lookup("1500100001")
    println(seq)


  }

}
